package com.lry.zk;


import com.lry.Callback;
import com.lry.RegisterType;
import com.lry.RegistryService;
import com.lry.spi.LoadLevel;
import com.lry.util.StringUtils;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;

import static com.lry.common.CommonConstants.ROOT;
import static com.lry.common.CommonConstants.SLASH;

@Slf4j
@LoadLevel(name = RegisterType.ZOOKEEPER)
public class ZookeeperRegister implements RegistryService {

    private AbstractZookeeperClient zkClient;

    private String root = SLASH + ROOT;

    public ZookeeperRegister(String address) {
        this.zkClient = new CuratorZookeeperClient(address);
    }


    @Override
    public void register(String path,String content) {
        if (!this.zkClient.existNode(root)) {
            zkClient.createPersistentData(root, "");
        }
        String comPath = root + path;
        if (zkClient.existNode(comPath)) {
            zkClient.deleteNode(comPath);
        }
        zkClient.createTemporaryData(comPath, content);
    }

    @Override
    public List<String> subscribe(Callback callback) {

        List<String> contents = new ArrayList<>();
        List<String> brokerNames = zkClient.getChildrenData(root);
        for (String brokerName : brokerNames) {
            String brokerNamePath = root + SLASH + brokerName;
            List<String> brokerIds = zkClient.getChildrenData(brokerNamePath);
            for (String brokerId : brokerIds) {
                String content = subscribe(brokerNamePath+SLASH+brokerId,callback);
                contents.add(content);
            }
        }
        return contents;
    }


    private String subscribe(String comPath,Callback callback) {
        String content = zkClient.getNodeData(comPath);
        watchNodeData(comPath,callback);
        return content;
    }

    private void watchNodeData(String comPath,Callback callback){
        zkClient.watchNodeData(comPath, watchedEvent-> {
            String content = zkClient.getNodeData(comPath);
            callback.callback(content);
            //收到回调之后在注册一次监听，这样能保证一直都收到消息
            watchNodeData(comPath,callback);
        });
    }
}
